﻿#include <stdio.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <fcntl.h> 
#include <stdlib.h> 
#include <sys/ioctl.h>
 
 
#include "../../src_comm/my_mqtt.h"

#include "../../../include/common/base.h"

#include "../../../include/rtdbms/rtdbFun.h"
  
#include "../../../include/net/ZNet.h"

#include "../../../include/fore/rtdb_fun.h"

#include "../../../include/rapidjson/rapidjson.h" 
#include "../../../include/rapidjson/document.h"

#include "../../../include/rapidjson/writer.h"

#include "../../../include/rapidjson/stringbuffer.h"

#include "../../../include/rapidjson/prettywriter.h"

#include "../../../include/rapidjson/filereadstream.h"
#include "../../../include/rapidjson/filewritestream.h"

 
CMyMQTTAsync	        g_myMqttClient;
int myRecv_Msgarrvd(void *context, char *topicName, int topicLen, MQTTAsync_message *message); 

int  SyncDbProcessThread( ZtagAcceptParam * p);
void regNewDevs(CMyMQTTAsync * pCli, FORE_DEV_TYPE * pDevType, char* nodeId, char* chrstatus ) ;

void exit_process( int signo )
{ 
	 
	int pid = getpid();
	printf( "\n syncDbSvr exit signo=%d pid=%d\n", signo, pid );
	
	exit(0);
}


int myRecv_Msgarrvd(void *context, char *topicName, int topicLen, MQTTAsync_message *message)
{ 
	MQTTAsync client = (MQTTAsync)context;
 
	char *  msg_buf = new char[message->payloadlen+1];
	memset( msg_buf, 0, message->payloadlen+1);
	memcpy(msg_buf, message->payload, message->payloadlen); 
 
	base::my_printf("Message arrived\n");
	base::my_printf("     topic: %s\n", topicName);
	base::my_printf("   message payloadlen=%d  : ", message->payloadlen);
  
	base::my_printf(" msg_buf = %s \n", msg_buf); 
 	
	//if( strcmp(topicName, reply_dev_type_topic) == 0 ) { 
	//	process fun(...)
	//}
	 

	delete [] msg_buf;
	msg_buf = NULL; 

	MQTTAsync_freeMessage(&message);
	MQTTAsync_free(topicName);  

	return 1;
}


int MY_PRINTF_TAG = 1;
int main(int argc, char *argv[])
{ 

	signal(SIGKILL , exit_process);
	signal(SIGINT  , exit_process);
	signal(SIGTERM , exit_process);	 
	signal(SIGQUIT , exit_process);	 


	printf( "\n--------syncDbSvr enter!!!\n" );
  	   
	if( !OpenDbms() ) {
		printf("open rtdbms error! \n");  
	}
	else{
		printf("open rtdbms sucess!\n"); 
	} 

	unsigned int ver = TDbFun::get_rtdb_Version();

	if( ver == CURRENT_RTDB_VERSION ) { 
	}
	else {
		printf("syncDbSvr: rtdb ver[%d] != cur rtdb version[%d]!\n", ver, CURRENT_RTDB_VERSION );   
		base::MySleep(2000);
		return 0;
	} 


	char mqttSvrIp[32] = "tcp://127.0.0.1:1883";   
	char  mqtt_client_id[64] ={0};
	sprintf(mqtt_client_id, "testClientID_%d",  getpid() ); 

	while(1)
	{
		if( g_myMqttClient.connect_mqtt(mqtt_client_id, mqttSvrIp, myRecv_Msgarrvd) )
		{
			base::my_printf("connect_mqtt  ok !\n"); 
  
			break;
		}  

		base::MySleep(1000);
	}


/////////////////////////////////////////


	char  key[64]={0};
	char nodeId[32] ={0};
	char meaName[32]={0};
	clock_t cb=clock();
	for( int i=5000; i<5100 ; i++ )
	{ 
		sprintf(nodeId, "GUID_%d", i+1 );
		sprintf(meaName, "meaName%d", i+1 );  //对应xml中的  PhV_phsA、PhV_phsB等 
		
		FORE_YC_POINT   yc_rec;
		strcpy(yc_rec.dev_nodeID, nodeId);
		strcpy(yc_rec.name, meaName);
		
		//TRACE("InsertRec=%s ", nodeId );
		UpdateRec_stringkey_Ex(FORE_YC_POINT_NO, nodeId, (char*)&yc_rec, sizeof(FORE_YC_POINT));
		InsertRec_stringkey(FORE_YC_POINT_NO, nodeId, (char*)&yc_rec, sizeof(FORE_YC_POINT));
	}
	clock_t ce=clock();
	base::my_printf(" cpu_clock=%0.6fs \n ",  ((double)(ce-cb))/(double)CLOCKS_PER_SEC   ); 
	SaveRtdb();
 
	
	FORE_DEV_TYPE * pDevTypeRec = (FORE_DEV_TYPE*)FindRec_stringkey(FORE_DEV_TYPE_NO, "Multimeters");
	if( pDevTypeRec == NULL ) {
		base::my_printf(" NO FIND  Multimeters IN DEV_TYPE \n" );
		return 0 ; 
	}

	for( int i=0; i<10; i++ )
	{
		char nodeId[64] ={0};
		sprintf(nodeId, "Multimeters%02d", i+1 );
		regNewDevs(&g_myMqttClient, pDevTypeRec, nodeId, "online");
	}
///////////////////////////////////////////


 
	ZTcpServerSocket   svrSocket;

	int scadaPort = 10135;
	if( svrSocket.Create(scadaPort, SyncDbProcessThread) ) 
	{
		base::my_printf("sync db thread socket create ok!!! \n");
	}
	else 
	{
		base::my_printf("sync db thread socket create error!!! \n");
	}
	 
	for( ; ; )
	{
		usleep( 30 * 1000 * 1000 );  // 30s
	}

	printf( "\n--------syncDbSvr exit...\n\n" );

	return 1;
}


void regNewDevs(CMyMQTTAsync * pCli, FORE_DEV_TYPE * pDevType, char* nodeId, char* chrstatus ) //online/offline
{
	if( pCli == NULL || nodeId == NULL || pDevType == NULL ) return;
	
	base::my_printf( "\n--------regNewDevs-------------------\n" ); 

	rapidjson::StringBuffer json; 
	rapidjson::Writer<rapidjson::StringBuffer> writer(json);  
	 
	
	writer.StartObject(); 
	writer.Key("token"); 
	writer.String("token580"); 
	writer.Key("timestamp");  
	writer.String( base::toTimeString(time(NULL)).c_str() ); 
	writer.Key("body"); 
	writer.StartObject();
	
	writer.Key("state"); 
	writer.String(chrstatus);
	writer.Key("event-time");  
	writer.String(base::toTimeString(time(NULL)).c_str()); 
		
	writer.EndObject();
	writer.EndObject();
	
	 
	 
	char   send_topic[128] = {0}; 
	//sprintf(send_topic, "testApp/notify/event/gwTerminal/status/%s/%s/%s/%s/%s/%s",  pDevType->manufacturerId, pDevType->manufacturerName,
	//		 pDevType->deviceType, pDevType->model, pDevType->protocolType, nodeId);
	sprintf(send_topic, "testApp/notify/event/gwTerminal/status/%s/%s", pDevType->deviceType, nodeId);
	               
	pCli->send_string(1, send_topic,  (char*)json.GetString() );

	base::my_printf((char*)json.GetString());
	base::my_printf("\n");	
}
 
